package drds.data_propagate.parse.binlog_event_position_manager;

import drds.data_propagate.binlog_event.secondary.ClientId;
import drds.data_propagate.binlog_event.secondary.position.BinLogEventPosition;
import drds.data_propagate.metadata.MetaDataManager;
import drds.data_propagate.parse.exception.ParseException;
import drds.data_propagate.store.EventUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.List;


public class MetaBinLogEventPositionManager extends AbstractBinLogEventPositionManager {

    private final static Logger logger = LoggerFactory.getLogger(MetaBinLogEventPositionManager.class);

    private final MetaDataManager metaDataManager;

    public MetaBinLogEventPositionManager(MetaDataManager metaDataManager) {
        if (metaDataManager == null) {
            throw new NullPointerException("null metaDataManager");
        }

        this.metaDataManager = metaDataManager;
    }

    @Override
    public void stop() {
        super.stop();

        if (metaDataManager.isStart()) {
            metaDataManager.stop();
        }
    }

    @Override
    public void start() {
        super.start();

        if (!metaDataManager.isStart()) {
            metaDataManager.start();
        }
    }

    @Override
    public BinLogEventPosition getLatestBinLogEventPosition(String taskId) {
        List<ClientId> clientIdList = metaDataManager.listAllSubscribeInfo(taskId);
        BinLogEventPosition result = null;
        if (!CollectionUtils.isEmpty(clientIdList)) {
            // 尝试找到一个最小的logPosition
            for (ClientId clientId : clientIdList) {
                BinLogEventPosition position = (BinLogEventPosition) metaDataManager.getPosition(clientId);
                if (position == null) {
                    continue;
                }

                if (result == null) {
                    result = position;
                } else {
                    result = EventUtils.min(result, position);
                }
            }
        }

        return result;
    }

    @Override
    public void persistBinLogEventPosition(String taskId, BinLogEventPosition binLogEventPosition) throws ParseException {
        // do nothing
        logger.info("persist BinlogEventPosition:{}", taskId, binLogEventPosition);
    }
}
